package com.tgy.kafka.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

/***
 * @ClassName: KafkaErrorListener
 * @Description: kafka错误监听
 * @Auther: tanggy3
 * @Date: 20:01 2022/6/23
 * @version : V1.0
 */
@Component
@Slf4j
public class KafkaErrorListener {
    @Bean("consumerAwareErrorHandler")
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
                log.error("consumerAwareErrorHandler receive: " + message.getPayload().toString(), e);
                return null;
            }
        };
    }
}
